package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/1 18:59
 */
public class DWD_11_DwdToolCouponGetApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_11_DwdToolCouponGetApp().init("DwdToolCouponGetApp", 1, 3011, "DwdToolCouponGetApp");
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        /*
            每领取一张优惠券, 表中会插入一条数据
                所以过滤insert数据即可
                
                
                coupon_use
         */
        
        // TODO 3. 从 Kafka 读取业务数据，封装为 Flink SQL 表
        tEnv.executeSql(
            "create table `topic_db`( " +
                "`database` string, " +
                "`table` string, " +
                "`data` map<string, string>, " +
                "`type` string, " +
                "`ts` string " +
                ")" + SQLUtil.getKafkaDDL("dwd_tool_coupon_get", "ods_db"));
        
        // TODO 4. 读取优惠券领用数据，封装为表
        Table resultTable = tEnv.sqlQuery(
            "select " +
                "data['id'], " +
                "data['coupon_id'], " +
                "data['user_id'], " +
                "date_format(data['get_time'],'yyyy-MM-dd') date_id, " +
                "data['get_time'], " +
                "ts " +
                "from topic_db " +
                "where `table` = 'coupon_use' " +
                "and `type` = 'insert' ");
        tEnv.createTemporaryView("result_table", resultTable);
        
        // TODO 5. 建立 Upsert-Kafka dwd_tool_coupon_get 表
        tEnv.executeSql("create table dwd_tool_coupon_get ( " +
                            "id string, " +
                            "coupon_id string, " +
                            "user_id string, " +
                            "date_id string, " +
                            "get_time string, " +
                            "ts string, " +
                            "primary key(id) not enforced " +
                            ")" + SQLUtil.getUpsertKafkaDDL("dwd_tool_coupon_get"));
        
        
        
        // TODO 6. 将数据写入 Upsert-Kafka 表
//        tEnv.executeSql("insert into dwd_tool_coupon_get select * from result_table");
    }
}
